package com.chiyoyo.rabbitmq.framework.utils;

import cn.hutool.core.util.ObjectUtil;
import com.chiyoyo.rabbitmq.framework.annotation.RabbitMq;
import com.chiyoyo.rabbitmq.framework.consumer.AbstractConsumer;
import com.chiyoyo.rabbitmq.framework.dto.MessageQueueDetail;
import com.chiyoyo.rabbitmq.framework.dto.RabbitMqModel;
import com.chiyoyo.rabbitmq.framework.init.RabbitMQInitConfig;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;

/**
 * @author chil
 */
@Slf4j
@Configuration
public class RabbitMqUtils {

    private final RabbitTemplate rabbitTemplate;
    private final CachingConnectionFactory cachingConnectionFactory;

    public RabbitMqUtils(RabbitTemplate rabbitTemplate, CachingConnectionFactory cachingConnectionFactory) {
        this.rabbitTemplate = rabbitTemplate;
        this.cachingConnectionFactory = cachingConnectionFactory;
    }

    /**
     * 发送消息到RabbitMQ
     *
     * @param mq 要发送的消息对象，实现了RabbitMqModel接口
     * @throws Exception 如果发送消息发生异常，则抛出该异常
     */
    public boolean send(RabbitMqModel<?> mq) throws Exception {
        return convertAndSend(mq, 0, false);
    }

    /**
     * 发送延缓信息
     *
     * @param mq    要发送的消息对象，实现了RabbitMqModel接口
     * @param delay 消息延迟发送的时间，单位为毫秒
     */
    public boolean sendDelay(RabbitMqModel<?> mq, long delay) {
        return convertAndSend(mq, delay, true);
    }

    /**
     * 发送消息，并设置消息过期时间
     *
     * @param mq         消息对象，实现了RabbitMqModel接口
     * @param expiration 消息过期时间，单位毫秒
     */
    public boolean sendExpiration(RabbitMqModel<?> mq, long expiration) {
        return convertAndSend(mq, expiration, false);
    }

    /**
     * 重置队列并发使用者
     *
     * @param queueName           队列名称
     * @param concurrentConsumers 并发使用者数量
     * @return 如果重置成功返回true，否则返回false
     * @throws IllegalStateException 如果并发使用者数量小于等于0
     */
    public boolean resetQueueConcurrentConsumers(String queueName, int concurrentConsumers) {
        Assert.state(concurrentConsumers > 0, "参数 'concurrentConsumers' 必须大于0.");
        DirectMessageListenerContainer container = (DirectMessageListenerContainer) findContainerByQueueName(queueName);
        if (ObjectUtil.isNotEmpty(container) && container.isActive() && container.isRunning()) {
            container.setConsumersPerQueue(concurrentConsumers);
            return true;
        }
        return false;
    }

    /**
     * 重启消息监听者
     *
     * @param queueName 队列名称
     * @return 如果重启成功返回true，否则返回false
     */
    public boolean restartMessageListener(String queueName) {
        if (ObjectUtil.isEmpty(queueName)) {
            log.debug("restartMessageListener 重启队列失败,传入队列名为空!");
            return false;
        }
        DirectMessageListenerContainer container = (DirectMessageListenerContainer) findContainerByQueueName(queueName);
        if (ObjectUtil.isEmpty(container)) {
            log.debug("restartMessageListener 停止队列失败,没有这个监听器");
            return false;
        }
        Assert.state(!container.isRunning(), String.format("消息队列%s对应的监听容器正在运行！", queueName));
        container.start();
        return true;
    }

    /**
     * 停止消息监听者。
     *
     * @param queueName 队列名称
     * @return 如果停止成功则返回true，否则返回false
     * @throws IllegalStateException 如果消息监听器未运行，则抛出此异常
     */
    public boolean stopMessageListener(String queueName) {
        if (ObjectUtil.isEmpty(queueName)) {
            log.debug("stopMessageListener 停止队列失败,传入队列名为空!");
            return false;
        }
        DirectMessageListenerContainer container = (DirectMessageListenerContainer) findContainerByQueueName(queueName);
        if (ObjectUtil.isEmpty(container)) {
            log.debug("stopMessageListener 停止队列失败,没有这个监听器");
            return false;
        }
        Assert.state(container.isRunning(), String.format("消息队列%s对应的监听容器未运行！", queueName));
        container.stop();
        return true;
    }

    /**
     * 将RabbitMqModel对象转换成消息并发送到RabbitMQ。
     *
     * @param rabbitMqModel 要转换和发送的RabbitMqModel对象，不能为空，否则抛出IllegalArgumentException异常
     * @param expiration    消息的过期时间，单位为毫秒
     * @param isDelayed     是否是延迟消息
     * @return 如果消息发送成功则返回true，否则返回false
     */
    private boolean convertAndSend(RabbitMqModel<?> rabbitMqModel, long expiration, boolean isDelayed) {
        if (rabbitMqModel == null || rabbitMqModel.getExchange() == null) {
            throw new IllegalArgumentException("RabbitMqModel 或 exchange 不能为空");
        }
        String originalVirtualHost = cachingConnectionFactory.getVirtualHost();
        boolean isUpdate = false;
        if (rabbitMqModel.getVirtualHost() != null && !rabbitMqModel.getVirtualHost().isEmpty()) {
            synchronized (cachingConnectionFactory) {
                // 添加同步块以确保线程安全
                cachingConnectionFactory.setVirtualHost(rabbitMqModel.getVirtualHost());
            }
            isUpdate = true;
        }
        try {
            rabbitTemplate.convertAndSend(
                    rabbitMqModel.getExchange(),
                    rabbitMqModel.getRoutingKey(),
                    // 判断是否只发送消息体
                    rabbitMqModel.getSendOnlyBody() ? rabbitMqModel.getBody() : rabbitMqModel,
                    msg -> configureMessage(msg, rabbitMqModel, expiration, isDelayed), new CorrelationData());
        } finally {
            if (isUpdate) {
                synchronized (cachingConnectionFactory) {
                    // 恢复原始的 virtual host
                    cachingConnectionFactory.setVirtualHost(originalVirtualHost);
                }
            }
        }
        return true;
    }

    /**
     * 配置消息属性
     *
     * @param msg           消息对象
     * @param rabbitMqModel RabbitMQ消息模型
     * @param expiration    消息过期时间
     * @param isDelayed     是否为延迟消息
     * @return 配置后的消息对象
     */
    private Message configureMessage(Message msg, RabbitMqModel<?> rabbitMqModel, long expiration, boolean isDelayed) {
        MessageProperties messageProperties = msg.getMessageProperties();
        // 如果 RabbitMqModel 中指定了内容类型，则设置消息的内容类型
        if (ObjectUtil.isNotEmpty(rabbitMqModel.getContentType())) {
            messageProperties.setContentType(rabbitMqModel.getContentType());
        }
        // 根据 RabbitMqModel 中的配置，设置消息的持久化模式
        messageProperties.setDeliveryMode(rabbitMqModel.getMessagePersistent() ? MessageDeliveryMode.PERSISTENT : MessageDeliveryMode.NON_PERSISTENT);

        if (expiration > 0) {
            if (ObjectUtil.isNotEmpty(rabbitMqModel.getHeaders())) {
                // 头信息添加到消息属性中
                rabbitMqModel.getHeaders().forEach(messageProperties::setHeader);
            }
            if (isDelayed) {
                // 如果是延迟消息，设置消息的延迟时间
                messageProperties.setHeader("x-delay", expiration);
            } else {
                // 否则，设置消息的过期时间
                messageProperties.setExpiration(String.valueOf(expiration));
            }
        }
        return msg;
    }

    /**
     * 获取所有队列消息监听容器列表
     *
     * @return 返回一个包含所有队列消息监听容器的集合
     */
    public Collection<AbstractMessageListenerContainer> getAllQueueContainerList() {
        return RabbitMQInitConfig.ALL_QUEUE_CONTAINER_MAP.values();
    }

    /**
     * 获取所有消息队列详情列表
     *
     * @return 消息队列详情列表
     */
    public List<MessageQueueDetail> statAllMessageQueueDetail() {
        List<MessageQueueDetail> queueDetailList = new ArrayList<>();
        RabbitMQInitConfig.ALL_QUEUE_CONTAINER_MAP.forEach((queueName, container) -> {
            queueDetailList.add(new MessageQueueDetail(queueName, container));
        });

        return queueDetailList;
    }

    /**
     * 根据队列名称查找消息监听容器
     *
     * @param queueName 队列名称
     * @return 对应的消息监听容器，若未找到则返回null
     */
    public AbstractMessageListenerContainer findContainerByQueueName(String queueName) {
        return RabbitMQInitConfig.ALL_QUEUE_CONTAINER_MAP.get(queueName);
    }

    /**
     * 获取指定队列名的DirectMessageListenerContainer容器
     *
     * @param queueName 队列名
     * @return 返回DirectMessageListenerContainer容器
     */
    private DirectMessageListenerContainer getContainer(String queueName) {
        return (DirectMessageListenerContainer) findContainerByQueueName(queueName);
    }

    /**
     * 手动拉去消息消费,根据队列名找到对应消费器消费
     *
     * @param transactional 是否开启事务
     * @param queueName     队列名
     */
    public void handleQueueMessageManually(boolean transactional, String queueName) throws Exception {
        Channel channel = rabbitTemplate.getConnectionFactory().createConnection().createChannel(transactional);
        GetResponse response = channel.basicGet(queueName, false);
        if (response == null) {
            log.debug("没有从{}队列中获取到消息", queueName);
            return;
        }
        Map<String, Object> annotationMap = SpringUtil.getBeansWithAnnotation(RabbitMq.class);
        for (Object obj : annotationMap.values()) {
            RabbitMq rabbitMq = AnnotationUtils.findAnnotation(obj.getClass(), RabbitMq.class);
            if (!queueName.equals(rabbitMq.queue())) {
                continue;
            }
            log.debug("从{}队列找到消费者类:{}", queueName, obj.getClass().getSimpleName());
            AbstractConsumer abstractConsumer = (AbstractConsumer) obj;
            abstractConsumer.onMessage(new Message(response.getBody(), new MessageProperties()), channel);
        }
    }
}
